package br.unicamp.mc715.middleware;

import java.io.Serializable;
import java.util.Timer;
import java.util.TimerTask;
import java.util.Vector;
import java.util.concurrent.PriorityBlockingQueue;

import br.unicamp.mc715.bank.BankOperation;
import br.unicamp.mc715.bank.BankOperation.BankOperationAck;
import br.unicamp.mc715.bank.BankOperation.BankOperationEnquiry;
import br.unicamp.mc715.bank.BankOperation.BankOperationRequest;
import br.unicamp.mc715.bank.BankOperation.BankOperationResponse;
import br.unicamp.mc715.common.Clock;
import br.unicamp.mc715.common.MC715IOException;
import br.unicamp.mc715.common.MC715SerializationException;
import br.unicamp.mc715.transport.Message;
import br.unicamp.mc715.transport.Transport;
import br.unicamp.mc715.transport.TransportId;
import br.unicamp.mc715.transport.UDPTransport;

/**
 * Objeto que implementa uma camada intermediária entre o Banco {@link Bank} e o transporte {@link UDPTransport}
 * Ele armazena os buffers de entrada e saída e posteriormente implementará um sistema de ordenação de mensagens. 
 *
 */
public class Middle extends Thread
{
	//private static final long MIN_MSG_DELAY = 10*1000000; //10ms
	
	/** O transporte UDP usado para a comunicação com outras máquinas. */
	private Transport transport;
	/** Numero de contas */
	private int numberOfAccounts;
	/** Numero de replicas do servico */
	private int numberOfInstances;
	/** Buffers de entrada, onde as operações ficam alojadas até serem requisitadas pela aplicação.
	 * E thread safe. */
	private PriorityBlockingQueue<BankOperation>[][] buffer;
	/** Seridores em execucao. */
	private Vector<TransportId> servers;
	/** Relogio local */
	private Clock clock;
	/** Verificacao de FIFO dos canais */
	//private Clock remoteClocks[];

	/** Contrutor da classe.
	 * @param numFilas Número máximo de filas que a aplicação pode exigir.
	 * @throws MC715IOException caso ocorra falha.
	 * @throws MC715SerializationException caso ocorra falha.
	 */
	@SuppressWarnings("unchecked")
	public Middle(int numContas, int numInstancias) throws MC715IOException, MC715SerializationException
	{
		this.transport = new UDPTransport();
		this.numberOfAccounts = numContas;
		this.numberOfInstances = numInstancias;
		//this.remoteClocks = new Clock[numInstancias];

		buffer = new PriorityBlockingQueue[this.numberOfAccounts][this.numberOfInstances];
		for(int j = 0; j < this.numberOfAccounts; j++)
			for(int i = 0; i < this.numberOfInstances; i++)
				buffer[j][i] = new PriorityBlockingQueue<BankOperation>();

		servers = new Vector<TransportId>();
		
		this.clock = new Clock(transport.getId());
		
		/* Envia mensagem de inicio para todos */
		transport.sendMessage(new Controle('s', getTransportId()));
		servers.add(getTransportId());
		
		this.start();
		
		/* Espera todos os servidores carregarem */
		while(getNumReplicas() != numberOfInstances)
			try {
				Thread.sleep(25);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
			
		/* Instancia thread para verificar se existem filas vazias. Ela envia
		 * pedidos de ACK para as filas vazias.
		 */
		startEnquiries();
	}

	/** Atualiza o relogio local com uma nova requisição 
	 * @param newClock Clock usado para atualizacao.
	 */
	public void update(Clock newClock) {
		this.clock.tictac(newClock);
	}

	/** Verifica se existe operacao na conta especificada.
	 * @param account Fila da conta na qual deve ser buscada a operação.
	 * @return 'True' se existe operação na fila. 'False' caso contrário.
	 */
	public boolean hasOp(int account) 
	{
		if (numberOfInstances == 1)
			return buffer[account][0].size() > 0;

		for(int i = 1; i < numberOfInstances; i++)
			if(buffer[account][i].peek() == null)
				return false;
		return true;
	}
	
	/** Retorna a primeira operacao da conta especificada. Remove-a da fila.
	 * @param account Fila da conta na qual deve ser buscada a operação.
	 * @return Operação.
	 */
	public BankOperation getOp(int account)
	{
		PriorityBlockingQueue<BankOperation> currentQueue = buffer[account][0];
		BankOperation currentOp = currentQueue.peek();

		for(int i = 1; i < numberOfInstances; i++) 
		{
			BankOperation op2 = buffer[account][i].peek();
			if (currentOp == null) {
				currentOp = op2;
				currentQueue = buffer[account][i];
			} else if (op2.getClock().lt(currentOp.getClock())) {
				currentOp = op2;
				currentQueue = buffer[account][i];
			}
		}
		
		if (currentOp == null) 
			return null;
		
		//if ((System.nanoTime() - currentOp.queueTime) < MIN_MSG_DELAY)
		//	return null;
		
		// System.out.println("currentOp:"+currentOp);
		currentQueue.remove(currentOp);
		
		if(currentOp.getOperationType() == 'a')
			return null;
		else
			return currentOp;
	}


	public void startEnquiries() {
		new Timer().schedule(new TimerTask() {
			@Override
			public void run() {
				sendEnquiries();
			}
		}, 0, 50);
	}
	
	public void sendEnquiries() {
		for(int a = 0; a < numberOfAccounts; a++) {
			for(int i = 1; i < numberOfInstances; i++) {
				if(buffer[a][i].peek() == null)
				{
					try {
						transport.sendMessage(new BankOperationEnquiry(clock, servers.get(i), a));
						Thread.sleep(3);
					} catch (Exception e) {
						e.printStackTrace();
					}
				}
			}
		}
	}
	
	/** Registra uma nova operacao no buffer se já não existe.
	 * @param op Operacao
	 * @throws MC715IOException
	 * @throws MC715SerializationException
	 */
	public void regOp(BankOperation op) throws MC715IOException, MC715SerializationException
	{
		PriorityBlockingQueue<BankOperation> queue = buffer[op.getOperationAccount()][servers.indexOf(op.getServerId())];
		//op.queueTime = System.nanoTime();
		
		if(op.getOperationType() != 'a')	
			queue.offer(op);
		else
			/* Se for ack so coloca se fila estiver vazia */
			if(queue.size() == 0)
				queue.offer(op);
	}

	/** Envia uma resposta a uma operacao de balanco enviada pelo cliente.
	 * @param resp Resposta a operacao enviada pelo cliente.
	 * 
	 */
	public void sendResponse(BankOperationResponse resp) throws MC715IOException, MC715SerializationException
	{
		if(transport.getId().equals(resp.operation.getServerId()))
		{
			/** Send response back to client */
			transport.sendMessage(resp, resp.operation.clientId);
		}
	}

	/*public void updateRemoteClock(Clock clock, TransportId id) {
		int i = servers.indexOf(id);
		if (remoteClocks[i] != null)
			if (remoteClocks[i].gt(clock))
				System.err.println("\n\n\n==\nOrdem das mensagens foi trocada!\n==\n\n\n\n");
		remoteClocks[i] = clock;
	}*/
	
	/** Simplesmente adiciona qualquer mensagem recebida a sua devida fila, separada por servidores.
	 *  Garante a correta ordenacao das mensagens.
	 */
	public void run()
	{
		while(true)
		{
			Serializable payload;

			BankOperation op = null;
			Controle ctrl = null;

			try {
				Message msg = transport.receiveMessage();

				if(msg != null)
				{
					payload = msg.getPayload();

					/** Trata mensagens enviadas por outra replica. */
					if(payload instanceof BankOperation)
					{
						//System.out.println("payload=BankOperation");
						op = (BankOperation)payload;
						/** Atualiza relogio local */
						update(op.getClock());
						//updateRemoteClock(op.getClock(), msg.getSender());
						/** Coloca operacao na fila. */
						regOp(op);
					}
					/** Registra o recebimento de um ACK. */
					else if(payload instanceof BankOperationAck && !msg.getSender().equals(transport.getId()))
					{
						// System.out.println("payload=Ack");
						BankOperationAck ack = (BankOperationAck)payload; 
						
						/** Atualiza relogio local */
						update(ack.clock);
						//updateRemoteClock(ack.clock, msg.getSender()); //FIXME
						
						/* ACK tem tipo 'a' */
						op = new BankOperation('a', ack.operationAccount, ack.clock);

						/** Coloca operacao na fila. */
						regOp(op);
					}
					/** Envia ACK no caso de um Enquiry. */
					else if(payload instanceof BankOperationEnquiry && !msg.getSender().equals(transport.getId()))
					{
						// System.out.println("payload=Enquiry");
						BankOperationEnquiry enq = (BankOperationEnquiry)payload;
						
						/** Atualiza relogio local */
						update(enq.clock);
						
						if (enq.destId.equals(transport.getId()))
						{
							//updateRemoteClock(enq.clock, msg.getSender()); //FIXME
							/** Envia um ACK. */
							transport.sendMessage(new BankOperationAck(clock, enq.operationAccount));
						}
					}
					/** Trata mensagens enviadas por um cliente. */
					else if (payload instanceof BankOperationRequest)
					{
						//System.out.println("payload=Request");
						BankOperationRequest request = (BankOperationRequest)payload;

						/** Se identifica como recipiente da requisicao */
						//System.out.println("ClientID=" + request.operation.clientId);

						/** Incrementa relogio local*/
						clock.tictac();
						request.operation.setClock(clock);

						/** Re-Envia a operacao para todas as replicas */
						transport.sendMessage(request.operation);
					}
					else if(payload instanceof Controle && !msg.getSender().equals(transport.getId()))
					{
						/** Trata mensagens de controle. */
						ctrl = (Controle)payload;

						if(ctrl.getType() == 's')
						{
							servers.add(ctrl.getServerId());
							transport.sendMessage(new Controle('c', getTransportId()), msg.getSender());
						}
						else if(ctrl.getType() == 'c')
						{
							servers.add(ctrl.getServerId());
						}
					}
				}

			} catch (MC715IOException e) {
				e.printStackTrace();
			} catch (MC715SerializationException e) {
				e.printStackTrace();
			}
		}
	}
	
	/** Retorna o numero de replicas presentes */
	public int getNumReplicas() {
		return servers.size();
	}

	/**Retorna o ID do transporte*/
	public TransportId getTransportId() {
		return transport.getId();
	}
}
